package com.mango.ch09;

import java.io.IOException;
import java.util.ArrayList;
import java.util.HashMap;
import java.util.List;
import java.util.Map;
import java.util.Map.Entry;

import org.apache.hadoop.io.LongWritable;
import org.apache.hadoop.io.Text;
import org.apache.hadoop.mapreduce.Reducer;

public class GroomFriendReducer extends Reducer<LongWritable, MyTuple, Text, Text> {

	@Override
	protected void reduce(LongWritable key, Iterable<MyTuple> values, Context context)
			throws IOException, InterruptedException {
		// TODO Auto-generated method stub
		// key=person
		// values=可能的推荐好友 List<Tuple2<userID1,userID2>>
		// mutualFriends.key 是推荐的好友
		// mutualFriends.value是共同好友列表
		Map<Long, List<Integer>> mutualFriends = new HashMap<>();
		System.out.print("当前redue中 key=" + key.get());
		for (MyTuple t : values) {
			System.out.print("  values= " + t.toString() + " ");
			Long toUser = t.get_1();// 推荐的好友的ID
			Integer mutualFriend = t.get_2();// 共同好友ID
			boolean alreadyFriend = (mutualFriend == -1);
			if (mutualFriends.containsKey(toUser)) {
				// 如果本人和该推荐用户已经是直接好友 则不用统计共同好友列表设置为null
				if (alreadyFriend)
					mutualFriends.put(toUser, null);
				// 否则将此人加入共同好友列表
				else if (mutualFriends.get(toUser) != null)
					mutualFriends.get(toUser).add(mutualFriend);
			} else {
				// 如果推荐好友集合中不存在此推荐用户,
				// 且已经是直接好友则不用添加
				if (alreadyFriend)
					mutualFriends.put(toUser, null);
				else {
					// 否则添加进集合中
					List<Integer> list = new ArrayList<>();
					list.add(mutualFriend);
					mutualFriends.put(toUser, list);
				}
			}
		}
		String reducerOutput = buildOutput(mutualFriends);
		context.write(new Text(key + ""), new Text(reducerOutput));
	}

	private String buildOutput(Map<Long, List<Integer>> mutualFriends) {
		// TODO Auto-generated method stub
		String output = "";
		for (Entry<Long, List<Integer>> entry : mutualFriends.entrySet()) {
			List<Integer> v = entry.getValue();
			String k = entry.getKey() + "";
			if (v==null)
				v=new ArrayList<>();
			output += k + "(" + v.size() + ":" + v.toString() + "), ";
		}
		return output;
	}
}
